package db.app

import com.cw.realtime.common.base.FlinkRunner.{RunnerConfig, SqlRunnerConfig, createTopicDbTable, runSql}
import com.cw.realtime.common.constant.Constant._
import com.cw.realtime.common.util.SqlUtil
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

import java.time.Duration

object DwdTradeOrderCancelDetail {


  def main(args: Array[String]): Unit = {

    implicit val config = SqlRunnerConfig(TOPIC_DWD_TRADE_ORDER_CANCEL, 10025)
    runSql { (_, tableEnv) =>
      tableEnv
        .getConfig
        .setIdleStateRetention(Duration.ofSeconds(15 * 60 + 5)) // 设定状态保留时间，等于订单开启时间 + 最大延迟

      createTopicDbTable(tableEnv)

      val orderCancel =  filterOrderCancel(tableEnv) // 从topic_db中筛选取消的订单
      tableEnv.createTemporaryView("order_cancel", orderCancel)

      readOrderDetailTable(tableEnv) // 读取dwd层订单明细

      createSinkTable(tableEnv)

      val joined = joinTable(tableEnv)
      tableEnv.createTemporaryView("joined", joined)
      joined.insertInto(TOPIC_DWD_TRADE_ORDER_CANCEL).execute()


    }


  }




def createSinkTable(tableEnv: StreamTableEnvironment) = {
  tableEnv.executeSql(
    s"""
create table $TOPIC_DWD_TRADE_ORDER_CANCEL (
  id   string,
  user_id   string,
  province_id   string,
  order_id   string,
  sku_id   string,
  sku_name   string,
  activity_id   string,
  activity_rule_id   string,
  coupon_id   string,
  order_price   string,
  sku_num   string,
  create_time   string,
  split_total_amount   string,
  split_activity_amount   string,
  split_coupon_amount   string,
  order_cancel_date_id   string,
  operate_time   string,
  ts   bigint
) ${SqlUtil.getKafkaSinkSql(TOPIC_DWD_TRADE_ORDER_CANCEL)}
    """
  )


}


  def readOrderDetailTable(tableEnv: StreamTableEnvironment) = {
    tableEnv.executeSql(
      s"""
create table $TOPIC_DWD_TRADE_ORDER_DETAIL (
  id                string,
  user_id           string,
  province_id       string,
  order_id          string,
  sku_id            string,
  activity_id       string,
  activity_rule_id  string,
  coupon_id         string,
  sku_name          string,
  order_price       string,
  sku_num           string,
  create_time       string,
  split_total_amount        string,
  split_activity_amount     string,
  split_coupon_amount       string,
  ts                bigint
) ${SqlUtil.getKafkaSourceSql(TOPIC_DWD_TRADE_ORDER_DETAIL, TOPIC_DWD_TRADE_ORDER_DETAIL)}
        """)
  }


  def filterOrderCancel(tableEnv: StreamTableEnvironment) = {
    tableEnv.sqlQuery(
      s"""
select
  `data`['id'] id,
  `data`['operate_time'] operate_time,
  `ts`
from topic_db
where
  `database` = 'gmall'
  and `table` = 'order_info'
  and `type` = 'update'
  and `old`['order_status'] is not null
  and `old`['order_status'] = '1001'
  and `data`['order_status'] = '1003'
      """
    )
//    tableEnv.executeSql("select * from order_cancel").print()

  }



    def joinTable(tableEnv: StreamTableEnvironment) = {

      tableEnv.sqlQuery(
        s"""
select
  od.id id,
  user_id,
  province_id,
  order_id,
  sku_id,
  sku_name,
  activity_id,
  activity_rule_id,
  coupon_id,
  order_price,
  sku_num,
  create_time,
  split_total_amount,
  split_activity_amount,
  split_coupon_amount,
  date_format(oc.operate_time, 'yyyy-MM-dd') order_cancel_date_id,
  oc.operate_time operate_time,
  oc.ts ts
  from ${TOPIC_DWD_TRADE_ORDER_DETAIL} od
  join order_cancel oc on
  od.order_id = oc.id
        """
      )

    }



}
